package com.king.app.dwd

import com.alibaba.fastjson.{JSON, JSONObject}
import com.king.util.MyKafkaUtil
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Author: KingWang
 * @Date: 2022/1/7  
 * @Desc:
 **/
object BaseLogApp {

  def main(args: Array[String]): Unit = {

    //1. 获取环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2.设置ck
//    env.setStateBackend(new FsStateBackend("hdfs://hadoop200:8020/flink/base_log_app/ck"))
//    env.enableCheckpointing(300000, CheckpointingMode.EXACTLY_ONCE)  //5分钟一次ck
//    env.getCheckpointConfig.setCheckpointTimeout(60000L)
//    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
//    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(300000L)


    //3. 添加kafka消费者
    val topic = "ods_base_log"
    val groupId = "ods_db_group"
    val kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic,groupId))


    //4. 将消费者数据解析为对象,并且将异常信息输出到侧输出流
    val dirtyTag = new OutputTag[String]("dirty")


    val jsonObjDS = kafkaDS.process(new ProcessFunction[String,JSONObject] {
      override def processElement(value: String, ctx: ProcessFunction[String, JSONObject]#Context, out: Collector[JSONObject]): Unit = {
        try{
          out.collect(JSON.parseObject(value))
        }catch{
          case e:Exception => ctx.output(dirtyTag,value)
        }
      }
    })


    //脏数据打印
    val dirtyDS = jsonObjDS.getSideOutput(dirtyTag)
    dirtyDS.print("dirty>>>>>>>")

    //5. 通过状态存储，标记是否新老用户
//    val jsonObjWithNewDS = jsonObjDS.keyBy(x=> x.getJSONObject("common").getString("mid"))  //这样直接写会报错
    val jsonObjWithNewDS = jsonObjDS.keyBy[String](new KeySelector[JSONObject,String] {
              override def getKey(value: JSONObject): String = value.getJSONObject("common").getString("mid")
        })
        .map(new RichMapFunction[JSONObject,JSONObject] {

          //第一种方法： 使用lazy功能先声明，在使用时再初始化，让该变量成为函数内全局变量
//          lazy val userState:ValueState[String] = getRuntimeContext
//            .getState[String](new ValueStateDescriptor[String]("userState", classOf[String]))

          //第二种方法： 定义可变量var，然后成为函数内全部变量
          var userState:ValueState[String] = _
          override def open(parameters: Configuration): Unit = {
            userState = getRuntimeContext.getState(new ValueStateDescriptor[String]("userState",classOf[String]))
          }

          override def map(value: JSONObject): JSONObject = {

            //获取is_new标记
            val is_new = value.getJSONObject("common").getString("is_new")

            //判断is_new是否为"1"，如果为1，则更新状态为新用户1,否则更新为0
            if("1".equals(is_new)){

              val tmp_userState = userState.value()
              if(null != tmp_userState) {
                //如果不为空，则表示有状态，只需要修改状态为0，表示老用户
                userState.update("0")
                value.getJSONObject("common").put("is_new","0")
              }else {
                //如果为空，则还没保存状态，则更新状态为1，表示新用户
                userState.update("1")
              }
            }
            value
          }
        })



    //6. 分流： 页面日志：主流， 曝光日志、启动日志：侧输出流
    val startTag = new OutputTag[String]("start")
    val displayTag = new OutputTag[String]("display")
    val pageDS = jsonObjWithNewDS.process(new ProcessFunction[JSONObject,String] {

      override def processElement(value: JSONObject, ctx: ProcessFunction[JSONObject, String]#Context, out: Collector[String]): Unit = {

        //获取自动日志自动
        val start = value.getString("start")
        if(null != start && start.length > 0){

          //1）将启动日志输出到侧输出流
          ctx.output(startTag,value.toJSONString)

        }else{
          //2)将页面数据写入主流
          out.collect(value.toJSONString)

          //获取曝光日志信息
          val displays = value.getJSONArray("displays")
          if(null != displays && displays.size() > 0){

            val pageId = value.getJSONObject("page").getString("page_id")

           (0 until displays.size()).map{displays.getJSONObject}.map{obj =>
             obj.put("page_id",pageId)   //添加字段页面Id
             //3) 将曝光数据展开的记录每一条写入测输出流
             ctx.output(displayTag, obj.toJSONString)
           }
          }
        }
      }
    })

    //7. 提取侧输出流
    val startDS = pageDS.getSideOutput(startTag)
    val displayDS = pageDS.getSideOutput(displayTag)

    //7. 将不同的流推送到kafka中不同的Topic
    startDS.print("start>>>>>>>>>")
    displayDS.print("display>>>>>>>>>")
    pageDS.print("page>>>>>>>>>>>>>")

    startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"))
    displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"))
    pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"))


    //8. 启动任务
    env.execute("base_log_app")

  }

}
